package com.st.kafkademon;

import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Properties;

import kafka.consumer.Consumer;
import kafka.consumer.ConsumerConfig;
import kafka.consumer.ConsumerIterator;
import kafka.consumer.KafkaStream;
import kafka.javaapi.consumer.ConsumerConnector;
import kafka.message.MessageAndMetadata;

/**
 * @author wuxianwei on 15/11/19.
 */
public class KafkaConsumer extends Thread {
  private final ConsumerConnector consumer;
  private final String topic;

  public KafkaConsumer(String topic) {
    Properties props = new Properties();
    props.put("zookeeper.connect", KafkaProperties.zkConnect);
    props.put("group.id", KafkaProperties.groupId);
    props.put("zookeeper.session.timeout.ms", "40000");
    props.put("zookeeper.sync.time.ms", "200");
    props.put("auto.commit.interval.ms", "1000");
    ConsumerConfig cusConf = new ConsumerConfig(props);
    this.consumer = Consumer.createJavaConsumerConnector(cusConf);
    this.topic = topic;
  }

  @Override
  public void run() {
    Map<String, Integer> topicCountMap = new HashMap<>();
    topicCountMap.put(topic, 1);//1原是 new Integer(1)
    Map<String, List<KafkaStream<byte[], byte[]>>> consumerMap =
            consumer.createMessageStreams(topicCountMap);
    KafkaStream<byte[], byte[]> stream = consumerMap.get(topic).get(0);//0
    //这里居然是阻塞的......
    for (MessageAndMetadata<byte[], byte[]> aStream : stream) {
      System.out.println("receive：" + new String(aStream.message()));
    }

    /**
     * 阻塞的 这么写比较好
     ConsumerIterator<byte[], byte[]> it = stream.iterator();
     while (it.hasNext()) {
     System.out.println("receive：" + new String(it.next().message()));
     try {
     sleep(3000);
     } catch (InterruptedException e) {
     e.printStackTrace();
     }
     }
     */
  }
}
